package club.blake.redisServer;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * Redis 服务器 模拟类
 * 实现了以下功能：
 * 1，redis 数据库的数据结构。-16个db，渐进式 rehash，redis客户端。
 * 2，redis 是怎么用单线程 处理客户端请求 和后台任务的流程。
 * 3，redis 时间事件和文件事件，及怎么使用nio的。
 * 4, resp 协议的处理。
 * 5，输入输出缓冲区。
 * 6，最后实现 auth , select , get ,set 命令
 *
 * ==目的== ：
 * 1，不是为了实现一个 Redis 服务器。
 * 2，是把我自己花时间研究和学习 redis的总结，分享给大家。
 *    为什么用java写，因为便于大家理解，redis源码是c 语言写的。我直接分享效果可能不是很好
 */
public class RedisServer {
  /**
   * redis 数据库，默认 16 个
   */
  static RedisDB[] redisDb;

  /**
   * 是否关闭服务
   */
  static boolean shutdown = false;

  /**
   * 每秒执行的次数(执行频率)
   */
  static int hz = 10;
  // 客户端连接集合
  static List<RedisClient> clients = new ArrayList<>();


  static ServerSocketChannel serverSocketChannel;
  static Selector selector;
  static HashMap<SelectionKey, RedisClient> clientMap =
      new HashMap<>();

  static class RedisDB {
    /**
     * 数据库键空间，保存了数据库中的所有键值对 key = keyName, value = "value"
     */
    Dict dict = new Dict();

    /**
     * 键的过期时间，保存了所有键的过期时间 key = keyName, value = "expireTime"
     */
    Dict expires = new Dict();

    public Object get(String key) {
      return dict.get(key);
    }

    public void set(String key, Object value) {
      dict.set(key, value);
    }
  }


  //模拟redis的数据结构
  static class Dict {
    // 一个字典使用两个哈希表，这样做的目的是为了实现渐进式 rehash
    // 一般使用的是 0号哈希表 ht[0]为哈希表，ht[1]为哈希表
    Hashtable[] ht = new Hashtable[2];
    {
      for (int i = 0; i < 2; i++) {
        ht[i] = new Hashtable();
      }
    }

    //是否正在渐进式 rehash
    int rehashidx = -1; //-1表示没有进行渐进式 rehash，否则表示进行渐进式 rehash

    public Object get(String key) {
      if (rehashidx == 1) {
        // 正在进行渐进式 rehash，则需要判断ht[0]和ht[1]
        Object o = ht[0].get(key);
        if (o != null) {
          return o;
        } else {
          return ht[1].get(key);
        }
      } else {
        return ht[0].get(key);
      }
    }

    public void set(String key, Object value) {
      if (rehashidx == -1) {
        ht[0].put(key, value); //没有进行渐进式 rehash，则直接放入ht[0]中
      } else {
        ht[1].put(key, value); //进行渐进式 rehash，则放入ht[1]中
      }
    }
  }

  static class RedisClient {
    /**
     * 当前客户端正在使用的数据库
     */
    RedisDB selectedDb;

    byte[] queryBuf = new byte[1024]; //输入缓冲区
    int queryBufLen; //输入缓冲区长度

    byte[] outBuf = new byte[1024]; //输出缓冲区
    int outBufLen; //输出缓冲区长度


    // ===== 非redis数据结构 =======
    Object retValue; //返回值
    SocketChannel channel; //客户端连接

    boolean read; //是否可读
    boolean write; //是否可写
    boolean accept; //是否可接受连接

    // 动态扩容逻辑
    private void ensureQueryBufCapacity(int required) {
      if (queryBufLen + required > queryBuf.length) {
        // 按需扩容（建议 2 倍扩容策略）
        int newCapacity = Math.max(queryBuf.length * 2, queryBufLen + required);
        queryBuf = Arrays.copyOf(queryBuf, newCapacity);
      }
    }

    // 将 ByteBuffer 中的数据追加到 queryBuf 中
    public void appendToQueryBuf(ByteBuffer buffer) {
      buffer.flip(); // 切换到读模式
      try {
        // 直接复制全部可读数据
        int bytesToCopy = buffer.remaining();
        ensureQueryBufCapacity(bytesToCopy); // 必须保证容量足够
        buffer.get(queryBuf, 0, bytesToCopy);
        queryBufLen = bytesToCopy;
      } finally {
        // 无论是否成功，必须清空缓冲区[1](@ref)
        buffer.clear();
      }
    }
  }

  /**
   * 初始化服务器
   */
  public static void initSever() throws IOException {
    /**
     * 从配置文件中读取配置信息
     */
    int dbCount = 16;
    int port = 6379;

    redisDb = new RedisDB[dbCount];
    for (int i = 0; i < dbCount; i++) {
      redisDb[i] = new RedisDB();
    }
    serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.configureBlocking(false);
    serverSocketChannel.socket().bind(new InetSocketAddress(port));
    selector = Selector.open();
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  }


  /**
   * 函数入口
   */
  public static void main(String[] args) throws Exception {
    System.out.println("init server");
    initSever();//初始化服务器
    /**
     * redis的时间事件默认间隔是 100ms。
     *
     * 假设第一次在35ms的时候，执行一次定时任务，
     * 那么下次定时任务执行的时间就是35ms + 100ms = 135ms
     * 再继续执行，直到下次定时任务执行的时间为135ms + 100ms = 235ms
     */
    long nextEventTime = System.currentTimeMillis();

    while (!shutdown) {
      long now = System.currentTimeMillis();
      long timeout = 0;
      if (now < nextEventTime) {
        /**
         * 还未到达 下次时间事件,
         * 计算超时时间= 下次时间事件 - 当前时间
         */
        timeout = nextEventTime - now;
      } else if (now > nextEventTime) {
        //到达了,设置下次时间事件
        nextEventTime = now + 100;
      }

      // 处理文件事件
      aeProcessEvents(timeout);

      //处理时间事件
      if (now >= nextEventTime) {
        //定时任务,默认每秒执行 hz(10) 次
        serverCron();
      }
    }
  }

  /**
   * 事件处理
   */
  public static void aeProcessEvents(long timeout) throws IOException {
    /**
     * 实际上 redis 是使用  epoll_wait 函数来监听事件的,
     * 且这里的休眠时间是计算出来的（计算规则 和 时间事件 有关系）
     *
     * 这里为了简化，直接使用 select 函数来监听事件
     */
    selector.select(timeout);

    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) {
      SelectionKey key = iterator.next();
      iterator.remove();
      System.out.println("处理事件" + key);

      if (key.isAcceptable()) {
        handleAccept(key);
      } else if (key.isReadable()) {
        handleRead(key);
      } else if (key.isWritable()) {
        handleWrite(key);
      }
    }
  }


  /**
   * 定时函数，每秒执行 hz 次,不进行设置默认是 每100ms执行一次
   */
  public static void serverCron() throws InterruptedException {
    //每秒执行hz次
    System.out.println("serverCron");
    //换行
    System.out.println();
    System.out.println();

    /**
     *  1, 关闭redis服务器
     *  2, 关闭超时客户端
     *  3, 对数据执行各种操作, 如过期键删除， 渐进式 rehash
     *  4， aof 持久化
     *  5, rdb 持久化
     *  6, 和主服务器同步数据, run_with_period(1000) - 1000ms一次
     *  7, 集群操作(集群模式)  gossip协议和其他节点通信，故障转移等  run_with_period(100) - 100ms一次
     *  8， sentinel模式，监控主从服务器状态，故障转移等  run_with_period(100) - 100ms一次
     *  9， 数据统计等，慢查询日志等
     */
    Thread.sleep(10);

  }

  /**
   * 处理连接事件
   */
  private static void handleAccept(SelectionKey key) throws IOException {
    RedisClient client = new RedisClient();
    SocketChannel clientChannel = serverSocketChannel.accept();
    if (clientChannel == null) return;

    System.out.println("接受连接" + clientChannel);
    clientChannel.configureBlocking(false); // 设置为非阻塞模式
    /** 注册读事件 */
    clientChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
    client.channel = clientChannel; //客户端绑定socket
    client.selectedDb = redisDb[0]; //默认选择第一个数据库
    clientMap.put(clientChannel.keyFor(selector), client);
    clients.add(client);
  }

  /**
   * 处理读事件
   */
  private static void handleRead(SelectionKey key) throws IOException {
    //拿到绑定客户端
    RedisClient redisClient = clientMap.get(key);
    if (redisClient == null) return;

    // 读取客户端数据
    SocketChannel socketChannel = (SocketChannel) key.channel();
    ByteBuffer buffer = (ByteBuffer) key.attachment();

    try {
      int bytesRead = socketChannel.read(buffer);
      if (bytesRead == -1) {
        closeClient(socketChannel, key, redisClient);
        return;
      }
    } catch (IOException e) {
      closeClient(socketChannel, key, redisClient);
    }
    System.out.println("读取数据:" + new String(buffer.array()));

    // 将数据追加到客户端的缓冲区中
    redisClient.appendToQueryBuf(buffer);
    processQueryBuffer(redisClient);
  }

  /**
   * 处理写事件
   */
  private static void handleWrite(SelectionKey key) throws IOException {
    RedisClient client = clientMap.get(key);
    if (client == null) return;

    if (!client.write) return;

    SocketChannel socketChannel = (SocketChannel) key.channel();

    // ============= 新增 RESP 协议封装逻辑 =============
    if (client.retValue == null) {
      client.outBuf = "$-1\r\n".getBytes(StandardCharsets.UTF_8);
    } else if (client.retValue instanceof String) {
      String rawValue = (String) client.retValue;
      // 判断响应类型（示例逻辑，需根据实际命令完善）
      if ("OK".equals(rawValue)) {
        client.outBuf = RespUtil.formatSimpleString(rawValue); // +OK\r\n
      } else {
        client.outBuf = RespUtil.formatBulkString(rawValue);    // $5\r\nhello\r\n
      }
    } else if (client.retValue instanceof Integer) {
      client.outBuf = RespUtil.formatInteger((Integer) client.retValue); // :42\r\n
    } else if (client.retValue instanceof Throwable) {
      client.outBuf = RespUtil.formatError(((Throwable) client.retValue).getMessage()); // -ERR...
    }
    // ================================================

    ByteBuffer buffer = ByteBuffer.wrap(client.outBuf);
    try {
      int totalWritten = 0;
      while (buffer.hasRemaining()) {
        int bytesWritten = socketChannel.write(buffer);
        if (bytesWritten <= 0) break;
        totalWritten += bytesWritten;
      }

      if (buffer.hasRemaining()) {
        // 分片写入优化
        if (totalWritten < client.outBuf.length) {
          client.outBuf = Arrays.copyOfRange(
              client.outBuf, totalWritten, client.outBuf.length
          );
        }
        key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
        client.write = true;
      } else {
        // 状态清零与事件切换
        //client.outBuf = null;
        client.retValue = null;
        // 取消写事件
        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
        //key.interestOps(SelectionKey.OP_READ);
        client.write = false;
      }
    } catch (Exception e) {
      System.out.println("write error: " + e.getMessage());
      closeClient(socketChannel, key, client); // 需实现资源释放
    }
  }

  /**
   * 处理客户端请求的缓冲区
   */
  private static void processQueryBuffer(RedisClient client) throws IOException {
    while (client.queryBufLen > 0) {
      RedisRequest cmd = new RedisRequest();
      int processed = RespUtil.parseCommand(client.queryBuf, 0, client.queryBufLen, cmd);

      if (processed > 0) {
        // 处理完整命令
        byte[] remaining = Arrays.copyOfRange(client.queryBuf, processed, client.queryBufLen);
        System.arraycopy(remaining, 0, client.queryBuf, 0, remaining.length);
        client.queryBufLen = remaining.length;

        Object result = doCommand(client, cmd);
        client.retValue = result;
        client.write = true;

        // 注册写事件
        SelectionKey key = client.channel.keyFor(selector);
        key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
      } else if (processed == 0) {
        break; // 数据不完整，等待下次读取
      } else {
        //closeClient(client.channel, client.channel.keyFor(selector), client);
        break;
      }
    }
  }

  /**
   * 关闭客户端连接
   */
  private static void closeClient(SocketChannel channel, SelectionKey key, RedisClient client) throws IOException {
    channel.close();
    clients.remove(client);
    clientMap.remove(key);
    System.out.println("关闭连接" + channel);
  }


  /**
   * 客户端请求信息
   */
  static class RedisRequest {
    String command;
    List<String> args;
  }


  public static Object doCommand(RedisClient redisClient, RedisRequest request) {
    System.out.println("handle command: " + request.command + " args: " + request.args);

    if ("get".equalsIgnoreCase(request.command)) {
      return redisClient.selectedDb.get(request.args.get(0));
    } else if ("set".equalsIgnoreCase(request.command)) {
      redisClient.selectedDb.set(request.args.get(0), request.args.get(1));
      return "OK";
    } else if ("select".equalsIgnoreCase(request.command)) {
      int dbIndex = Integer.parseInt(request.args.get(0));
      if (dbIndex < 0 || dbIndex >= redisDb.length) {
        return "ERR invalid DB index";
      }
      redisClient.selectedDb = redisDb[dbIndex];
      return "OK";
    } else if ("auth".equalsIgnoreCase(request.command)) {
      return "OK";
    }
    return "ERR unknown command '" + request.command + "'";
  }
}

